/*-------------------------------------------------------------------------
 *
 * mvdistinct.c
 *	  POSTGRES multivariate ndistinct coefficients
 *
 * Estimating number of groups in a combination of columns (e.g. for GROUP BY)
 * is tricky, and the estimation error is often significant.

 * The multivariate ndistinct coefficients address this by storing ndistinct
 * estimates for combinations of the user-specified columns.  So for example
 * given a statistics object on three columns (a,b,c), this module estimates
 * and stores n-distinct for (a,b), (a,c), (b,c) and (a,b,c).  The per-column
 * estimates are already available in pg_statistic.
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/statistics/mvdistinct.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <math.h>

#include "access/htup_details.h"
#include "catalog/pg_statistic_ext.h"
#include "catalog/pg_statistic_ext_data.h"
#include "lib/stringinfo.h"
#include "statistics/extended_stats_internal.h"
#include "statistics/statistics.h"
#include "utils/fmgrprotos.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"
#include "utils/typcache.h"

static double fc_ndistinct_for_combination(double fc_totalrows, StatsBuildData *fc_data,
										int fc_k, int *fc_combination);
static double fc_estimate_ndistinct(double fc_totalrows, int fc_numrows, int fc_d, int fc_f1);
static int	fc_n_choose_k(int fc_n, int fc_k);
static int	fc_num_combinations(int fc_n);

/* size of the struct header fields (magic, type, nitems) */
#define SizeOfHeader		(3 * sizeof(uint32))

/* size of a serialized ndistinct item (coefficient, natts, atts) */
#define SizeOfItem(natts) \
	(sizeof(double) + sizeof(int) + (natts) * sizeof(AttrNumber))

/* minimal size of a ndistinct item (with two attributes) */
#define MinSizeOfItem	SizeOfItem(2)

/* minimal size of mvndistinct, when all items are minimal */
#define MinSizeOfItems(nitems)	\
	(SizeOfHeader + (nitems) * MinSizeOfItem)

/* Combination generator API */

/* internal state for generator of k-combinations of n elements */
typedef struct CombinationGenerator
{
	int			k;				/* size of the combination */
	int			n;				/* total number of elements */
	int			current;		/* index of the next combination to return */
	int			ncombinations;	/* number of combinations (size of array) */
	int		   *combinations;	/* array of pre-built combinations */
} CombinationGenerator;

static CombinationGenerator *fc_generator_init(int fc_n, int fc_k);
static void fc_generator_free(CombinationGenerator *fc_state);
static int *fc_generator_next(CombinationGenerator *fc_state);
static void fc_generate_combinations(CombinationGenerator *fc_state);


/*
 * statext_ndistinct_build
 *		Compute ndistinct coefficient for the combination of attributes.
 *
 * This computes the ndistinct estimate using the same estimator used
 * in analyze.c and then computes the coefficient.
 *
 * To handle expressions easily, we treat them as system attributes with
 * negative attnums, and offset everything by number of expressions to
 * allow using Bitmapsets.
 */
MVNDistinct * statext_ndistinct_build(double fc_totalrows, StatsBuildData *fc_data)
{
	MVNDistinct *fc_result;
	int			fc_k;
	int			fc_itemcnt;
	int			fc_numattrs = fc_data->nattnums;
	int			fc_numcombs = fc_num_combinations(fc_numattrs);

	fc_result = palloc(offsetof(MVNDistinct, items) +
					fc_numcombs * sizeof(MVNDistinctItem));
	fc_result->magic = STATS_NDISTINCT_MAGIC;
	fc_result->type = STATS_NDISTINCT_TYPE_BASIC;
	fc_result->nitems = fc_numcombs;

	fc_itemcnt = 0;
	for (fc_k = 2; fc_k <= fc_numattrs; fc_k++)
	{
		int		   *fc_combination;
		CombinationGenerator *fc_generator;

		/* generate combinations of K out of N elements */
		fc_generator = fc_generator_init(fc_numattrs, fc_k);

		while ((fc_combination = fc_generator_next(fc_generator)))
		{
			MVNDistinctItem *fc_item = &fc_result->items[fc_itemcnt];
			int			fc_j;

			fc_item->attributes = palloc(sizeof(AttrNumber) * fc_k);
			fc_item->nattributes = fc_k;

			/* translate the indexes to attnums */
			for (fc_j = 0; fc_j < fc_k; fc_j++)
			{
				fc_item->attributes[fc_j] = fc_data->attnums[fc_combination[fc_j]];

				Assert(AttributeNumberIsValid(fc_item->attributes[fc_j]));
			}

			fc_item->ndistinct =
				fc_ndistinct_for_combination(fc_totalrows, fc_data, fc_k, fc_combination);

			fc_itemcnt++;
			Assert(fc_itemcnt <= fc_result->nitems);
		}

		fc_generator_free(fc_generator);
	}

	/* must consume exactly the whole output array */
	Assert(fc_itemcnt == fc_result->nitems);

	return fc_result;
}

/*
 * statext_ndistinct_load
 *		Load the ndistinct value for the indicated pg_statistic_ext tuple
 */
MVNDistinct * statext_ndistinct_load(Oid fc_mvoid, bool fc_inh)
{
	MVNDistinct *fc_result;
	bool		fc_isnull;
	Datum		fc_ndist;
	HeapTuple	fc_htup;

	fc_htup = SearchSysCache2(STATEXTDATASTXOID,
						   ObjectIdGetDatum(fc_mvoid), BoolGetDatum(fc_inh));
	if (!HeapTupleIsValid(fc_htup))
		elog(ERROR, "cache lookup failed for statistics object %u", fc_mvoid);

	fc_ndist = SysCacheGetAttr(STATEXTDATASTXOID, fc_htup,
							Anum_pg_statistic_ext_data_stxdndistinct, &fc_isnull);
	if (fc_isnull)
		elog(ERROR,
			 "requested statistics kind \"%c\" is not yet built for statistics object %u",
			 STATS_EXT_NDISTINCT, fc_mvoid);

	fc_result = statext_ndistinct_deserialize(DatumGetByteaPP(fc_ndist));

	ReleaseSysCache(fc_htup);

	return fc_result;
}

/*
 * statext_ndistinct_serialize
 *		serialize ndistinct to the on-disk bytea format
 */
bytea * statext_ndistinct_serialize(MVNDistinct *fc_ndistinct)
{
	int			fc_i;
	bytea	   *fc_output;
	char	   *fc_tmp;
	Size		fc_len;

	Assert(fc_ndistinct->magic == STATS_NDISTINCT_MAGIC);
	Assert(fc_ndistinct->type == STATS_NDISTINCT_TYPE_BASIC);

	/*
	 * Base size is size of scalar fields in the struct, plus one base struct
	 * for each item, including number of items for each.
	 */
	fc_len = VARHDRSZ + SizeOfHeader;

	/* and also include space for the actual attribute numbers */
	for (fc_i = 0; fc_i < fc_ndistinct->nitems; fc_i++)
	{
		int			fc_nmembers;

		fc_nmembers = fc_ndistinct->items[fc_i].nattributes;
		Assert(fc_nmembers >= 2);

		fc_len += SizeOfItem(fc_nmembers);
	}

	fc_output = (bytea *) palloc(fc_len);
	SET_VARSIZE(fc_output, fc_len);

	fc_tmp = VARDATA(fc_output);

	/* Store the base struct values (magic, type, nitems) */
	memcpy(fc_tmp, &fc_ndistinct->magic, sizeof(uint32));
	fc_tmp += sizeof(uint32);
	memcpy(fc_tmp, &fc_ndistinct->type, sizeof(uint32));
	fc_tmp += sizeof(uint32);
	memcpy(fc_tmp, &fc_ndistinct->nitems, sizeof(uint32));
	fc_tmp += sizeof(uint32);

	/*
	 * store number of attributes and attribute numbers for each entry
	 */
	for (fc_i = 0; fc_i < fc_ndistinct->nitems; fc_i++)
	{
		MVNDistinctItem fc_item = fc_ndistinct->items[fc_i];
		int			fc_nmembers = fc_item.nattributes;

		memcpy(fc_tmp, &fc_item.ndistinct, sizeof(double));
		fc_tmp += sizeof(double);
		memcpy(fc_tmp, &fc_nmembers, sizeof(int));
		fc_tmp += sizeof(int);

		memcpy(fc_tmp, fc_item.attributes, sizeof(AttrNumber) * fc_nmembers);
		fc_tmp += fc_nmembers * sizeof(AttrNumber);

		/* protect against overflows */
		Assert(fc_tmp <= ((char *) fc_output + fc_len));
	}

	/* check we used exactly the expected space */
	Assert(fc_tmp == ((char *) fc_output + fc_len));

	return fc_output;
}

/*
 * statext_ndistinct_deserialize
 *		Read an on-disk bytea format MVNDistinct to in-memory format
 */
MVNDistinct * statext_ndistinct_deserialize(bytea *fc_data)
{
	int			fc_i;
	Size		fc_minimum_size;
	MVNDistinct fc_ndist;
	MVNDistinct *fc_ndistinct;
	char	   *fc_tmp;

	if (fc_data == NULL)
		return NULL;

	/* we expect at least the basic fields of MVNDistinct struct */
	if (VARSIZE_ANY_EXHDR(fc_data) < SizeOfHeader)
		elog(ERROR, "invalid MVNDistinct size %zd (expected at least %zd)",
			 VARSIZE_ANY_EXHDR(fc_data), SizeOfHeader);

	/* initialize pointer to the data part (skip the varlena header) */
	fc_tmp = VARDATA_ANY(fc_data);

	/* read the header fields and perform basic sanity checks */
	memcpy(&fc_ndist.magic, fc_tmp, sizeof(uint32));
	fc_tmp += sizeof(uint32);
	memcpy(&fc_ndist.type, fc_tmp, sizeof(uint32));
	fc_tmp += sizeof(uint32);
	memcpy(&fc_ndist.nitems, fc_tmp, sizeof(uint32));
	fc_tmp += sizeof(uint32);

	if (fc_ndist.magic != STATS_NDISTINCT_MAGIC)
		elog(ERROR, "invalid ndistinct magic %08x (expected %08x)",
			 fc_ndist.magic, STATS_NDISTINCT_MAGIC);
	if (fc_ndist.type != STATS_NDISTINCT_TYPE_BASIC)
		elog(ERROR, "invalid ndistinct type %d (expected %d)",
			 fc_ndist.type, STATS_NDISTINCT_TYPE_BASIC);
	if (fc_ndist.nitems == 0)
		elog(ERROR, "invalid zero-length item array in MVNDistinct");

	/* what minimum bytea size do we expect for those parameters */
	fc_minimum_size = MinSizeOfItems(fc_ndist.nitems);
	if (VARSIZE_ANY_EXHDR(fc_data) < fc_minimum_size)
		elog(ERROR, "invalid MVNDistinct size %zd (expected at least %zd)",
			 VARSIZE_ANY_EXHDR(fc_data), fc_minimum_size);

	/*
	 * Allocate space for the ndistinct items (no space for each item's
	 * attnos: those live in bitmapsets allocated separately)
	 */
	fc_ndistinct = palloc0(MAXALIGN(offsetof(MVNDistinct, items)) +
						(fc_ndist.nitems * sizeof(MVNDistinctItem)));
	fc_ndistinct->magic = fc_ndist.magic;
	fc_ndistinct->type = fc_ndist.type;
	fc_ndistinct->nitems = fc_ndist.nitems;

	for (fc_i = 0; fc_i < fc_ndistinct->nitems; fc_i++)
	{
		MVNDistinctItem *fc_item = &fc_ndistinct->items[fc_i];

		/* ndistinct value */
		memcpy(&fc_item->ndistinct, fc_tmp, sizeof(double));
		fc_tmp += sizeof(double);

		/* 属性的数量 */  
		memcpy(&fc_item->nattributes, fc_tmp, sizeof(int));
		fc_tmp += sizeof(int);
		Assert((fc_item->nattributes >= 2) && (fc_item->nattributes <= STATS_MAX_DIMENSIONS));

		fc_item->attributes
			= (AttrNumber *) palloc(fc_item->nattributes * sizeof(AttrNumber));

		memcpy(fc_item->attributes, fc_tmp, sizeof(AttrNumber) * fc_item->nattributes);
		fc_tmp += sizeof(AttrNumber) * fc_item->nattributes;

		/* still within the bytea */
		Assert(fc_tmp <= ((char *) fc_data + VARSIZE_ANY(fc_data)));
	}

	/* we should have consumed the whole bytea exactly */
	Assert(fc_tmp == ((char *) fc_data + VARSIZE_ANY(fc_data)));

	return fc_ndistinct;
}

/*
 * pg_ndistinct_in
 *		input routine for type pg_ndistinct
 *
 * pg_ndistinct is real enough to be a table column, but it has no
 * operations of its own, and disallows input (just like pg_node_tree).
 */
Datum pg_ndistinct_in(PG_FUNCTION_ARGS)
{
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("cannot accept a value of type %s", "pg_ndistinct")));

	PG_RETURN_VOID();			/* 保持编译器安静 */
}

/*
 * pg_ndistinct
 *		output routine for type pg_ndistinct
 *
 * Produces a human-readable representation of the value.
 */
Datum pg_ndistinct_out(PG_FUNCTION_ARGS)
{
	bytea	   *fc_data = PG_GETARG_BYTEA_PP(0);
	MVNDistinct *fc_ndist = statext_ndistinct_deserialize(fc_data);
	int			fc_i;
	StringInfoData fc_str;

	initStringInfo(&fc_str);
	appendStringInfoChar(&fc_str, '{');

	for (fc_i = 0; fc_i < fc_ndist->nitems; fc_i++)
	{
		int			fc_j;
		MVNDistinctItem fc_item = fc_ndist->items[fc_i];

		if (fc_i > 0)
			appendStringInfoString(&fc_str, ", ");

		for (fc_j = 0; fc_j < fc_item.nattributes; fc_j++)
		{
			AttrNumber	fc_attnum = fc_item.attributes[fc_j];

			appendStringInfo(&fc_str, "%s%d", (fc_j == 0) ? "\"" : ", ", fc_attnum);
		}
		appendStringInfo(&fc_str, "\": %d", (int) fc_item.ndistinct);
	}

	appendStringInfoChar(&fc_str, '}');

	PG_RETURN_CSTRING(fc_str.data);
}

/*
 * pg_ndistinct_recv
 *		binary input routine for type pg_ndistinct
 */
Datum pg_ndistinct_recv(PG_FUNCTION_ARGS)
{
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("cannot accept a value of type %s", "pg_ndistinct")));

	PG_RETURN_VOID();			/* 保持编译器安静 */
}

/*
 * pg_ndistinct_send
 *		binary output routine for type pg_ndistinct
 *
 * n-distinct is serialized into a bytea value, so let's send that.
 */
Datum pg_ndistinct_send(PG_FUNCTION_ARGS)
{
	return byteasend(fcinfo);
}

/*
 * ndistinct_for_combination
 *		Estimates number of distinct values in a combination of columns.
 *
 * This uses the same ndistinct estimator as compute_scalar_stats() in
 * ANALYZE, i.e.,
 *		n*d / (n - f1 + f1*n/N)
 *
 * except that instead of values in a single column we are dealing with
 * combination of multiple columns.
 */
static double fc_ndistinct_for_combination(double fc_totalrows, StatsBuildData *fc_data,
						  int fc_k, int *fc_combination)
{
	int			fc_i,
				fc_j;
	int			fc_f1,
				fc_cnt,
				fc_d;
	bool	   *fc_isnull;
	Datum	   *fc_values;
	SortItem   *fc_items;
	MultiSortSupport fc_mss;
	int			fc_numrows = fc_data->numrows;

	fc_mss = multi_sort_init(fc_k);

	/*
	 * In order to determine the number of distinct elements, create separate
	 * values[]/isnull[] arrays with all the data we have, then sort them
	 * using the specified column combination as dimensions.  We could try to
	 * sort in place, but it'd probably be more complex and bug-prone.
	 */
	fc_items = (SortItem *) palloc(fc_numrows * sizeof(SortItem));
	fc_values = (Datum *) palloc0(sizeof(Datum) * fc_numrows * fc_k);
	fc_isnull = (bool *) palloc0(sizeof(bool) * fc_numrows * fc_k);

	for (fc_i = 0; fc_i < fc_numrows; fc_i++)
	{
		fc_items[fc_i].values = &fc_values[fc_i * fc_k];
		fc_items[fc_i].isnull = &fc_isnull[fc_i * fc_k];
	}

	/*
	 * For each dimension, set up sort-support and fill in the values from the
	 * sample data.
	 *
	 * We use the column data types' default sort operators and collations;
	 * perhaps at some point it'd be worth using column-specific collations?
	 */
	for (fc_i = 0; fc_i < fc_k; fc_i++)
	{
		Oid			fc_typid;
		TypeCacheEntry *fc_type;
		Oid			fc_collid = InvalidOid;
		VacAttrStats *fc_colstat = fc_data->stats[fc_combination[fc_i]];

		fc_typid = fc_colstat->attrtypid;
		fc_collid = fc_colstat->attrcollid;

		fc_type = lookup_type_cache(fc_typid, TYPECACHE_LT_OPR);
		if (fc_type->lt_opr == InvalidOid) /* 不应该发生 */
			elog(ERROR, "cache lookup failed for ordering operator for type %u",
				 fc_typid);

		/* prepare the sort function for this dimension */
		multi_sort_add_dimension(fc_mss, fc_i, fc_type->lt_opr, fc_collid);

		/* accumulate all the data for this dimension into the arrays */
		for (fc_j = 0; fc_j < fc_numrows; fc_j++)
		{
			fc_items[fc_j].values[fc_i] = fc_data->values[fc_combination[fc_i]][fc_j];
			fc_items[fc_j].isnull[fc_i] = fc_data->nulls[fc_combination[fc_i]][fc_j];
		}
	}

	/* We can sort the array now ... */
	qsort_interruptible((void *) fc_items, fc_numrows, sizeof(SortItem),
						multi_sort_compare, fc_mss);

	/* ... and count the number of distinct combinations */

	fc_f1 = 0;
	fc_cnt = 1;
	fc_d = 1;
	for (fc_i = 1; fc_i < fc_numrows; fc_i++)
	{
		if (multi_sort_compare(&fc_items[fc_i], &fc_items[fc_i - 1], fc_mss) != 0)
		{
			if (fc_cnt == 1)
				fc_f1 += 1;

			fc_d++;
			fc_cnt = 0;
		}

		fc_cnt += 1;
	}

	if (fc_cnt == 1)
		fc_f1 += 1;

	return fc_estimate_ndistinct(fc_totalrows, fc_numrows, fc_d, fc_f1);
}

/* The Duj1 estimator (already used in analyze.c). */
static double fc_estimate_ndistinct(double fc_totalrows, int fc_numrows, int fc_d, int fc_f1)
{
	double		fc_numer,
				fc_denom,
				fc_ndistinct;

	fc_numer = (double) fc_numrows * (double) fc_d;

	fc_denom = (double) (fc_numrows - fc_f1) +
		(double) fc_f1 * (double) fc_numrows / fc_totalrows;

	fc_ndistinct = fc_numer / fc_denom;

	/* 在发生舍入误差的情况下，限制在合理范围内 */
	if (fc_ndistinct < (double) fc_d)
		fc_ndistinct = (double) fc_d;

	if (fc_ndistinct > fc_totalrows)
		fc_ndistinct = fc_totalrows;

	return floor(fc_ndistinct + 0.5);
}

/*
 * n_choose_k
 *		computes binomial coefficients using an algorithm that is both
 *		efficient and prevents overflows
 */
static int fc_n_choose_k(int fc_n, int fc_k)
{
	int			fc_d,
				fc_r;

	Assert((fc_k > 0) && (fc_n >= fc_k));

	/* use symmetry of the binomial coefficients */
	fc_k = Min(fc_k, fc_n - fc_k);

	fc_r = 1;
	for (fc_d = 1; fc_d <= fc_k; ++fc_d)
	{
		fc_r *= fc_n--;
		fc_r /= fc_d;
	}

	return fc_r;
}

/*
 * num_combinations
 *		number of combinations, excluding single-value combinations
 */
static int fc_num_combinations(int fc_n)
{
	return (1 << fc_n) - (fc_n + 1);
}

/*
 * generator_init
 *		initialize the generator of combinations
 *
 * The generator produces combinations of K elements in the interval (0..N).
 * We prebuild all the combinations in this method, which is simpler than
 * generating them on the fly.
 */
static CombinationGenerator *
fc_generator_init(int fc_n, int fc_k)
{
	CombinationGenerator *fc_state;

	Assert((fc_n >= fc_k) && (fc_k > 0));

	/* allocate the generator state as a single chunk of memory */
	fc_state = (CombinationGenerator *) palloc(sizeof(CombinationGenerator));

	fc_state->ncombinations = fc_n_choose_k(fc_n, fc_k);

	/* pre-allocate space for all combinations */
	fc_state->combinations = (int *) palloc(sizeof(int) * fc_k * fc_state->ncombinations);

	fc_state->current = 0;
	fc_state->k = fc_k;
	fc_state->n = fc_n;

	/* now actually pre-generate all the combinations of K elements */
	fc_generate_combinations(fc_state);

	/* make sure we got the expected number of combinations */
	Assert(fc_state->current == fc_state->ncombinations);

	/* reset the number, so we start with the first one */
	fc_state->current = 0;

	return fc_state;
}

/*
 * generator_next
 *		returns the next combination from the prebuilt list
 *
 * Returns a combination of K array indexes (0 .. N), as specified to
 * generator_init), or NULL when there are no more combination.
 */
static int *
fc_generator_next(CombinationGenerator *fc_state)
{
	if (fc_state->current == fc_state->ncombinations)
		return NULL;

	return &fc_state->combinations[fc_state->k * fc_state->current++];
}

/*
 * generator_free
 *		free the internal state of the generator
 *
 * Releases the generator internal state (pre-built combinations).
 */
static void fc_generator_free(CombinationGenerator *fc_state)
{
	pfree(fc_state->combinations);
	pfree(fc_state);
}

/*
 * generate_combinations_recurse
 *		given a prefix, generate all possible combinations
 *
 * Given a prefix (first few elements of the combination), generate following
 * elements recursively. We generate the combinations in lexicographic order,
 * which eliminates permutations of the same combination.
 */
static void fc_generate_combinations_recurse(CombinationGenerator *fc_state,
							  int fc_index, int fc_start, int *fc_current)
{
	/* If we haven't filled all the elements, simply recurse. */
	if (fc_index < fc_state->k)
	{
		int			fc_i;

		/*
		 * The values have to be in ascending order, so make sure we start
		 * with the value passed by parameter.
		 */

		for (fc_i = fc_start; fc_i < fc_state->n; fc_i++)
		{
			fc_current[fc_index] = fc_i;
			fc_generate_combinations_recurse(fc_state, (fc_index + 1), (fc_i + 1), fc_current);
		}

		return;
	}
	else
	{
		/* we got a valid combination, add it to the array */
		memcpy(&fc_state->combinations[(fc_state->k * fc_state->current)],
			   fc_current, fc_state->k * sizeof(int));
		fc_state->current++;
	}
}

/*
 * generate_combinations
 *		generate all k-combinations of N elements
 */
static void fc_generate_combinations(CombinationGenerator *fc_state)
{
	int		   *fc_current = (int *) palloc0(sizeof(int) * fc_state->k);

	fc_generate_combinations_recurse(fc_state, 0, 0, fc_current);

	pfree(fc_current);
}
